package com.demo.rocketmq.spring.config;

import com.demo.rocketmq.annotation.MqConsumer;
import com.demo.rocketmq.spring.support.RocketMqListenerContainer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.MethodIntrospector;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.lang.NonNull;

import java.lang.reflect.Method;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 自定义mq container，获取使用@MqConsumer注解的Bean method对象，为每个topic生成一个consumer
 * 参考代码{@link org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration},{@link org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor#postProcessAfterInitialization}
 * @author cyj
  */
@Slf4j
@Configuration
public class RocketMqListenerContainerConfig implements ApplicationContextAware, BeanPostProcessor {

    private final Set<Class<?>> nonAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap<>(64));

    private ConfigurableApplicationContext applicationContext;

    private final RocketMQProperties rocketMQProperties;

    private final RocketMQMessageConverter rocketMQMessageConverter;

    private final RocketMQTemplate rocketMqTemplate;

    private final StandardEnvironment environment;

    public RocketMqListenerContainerConfig(RocketMQProperties rocketMqProperties,
                                           RocketMQMessageConverter rocketMqMessageConverter,
                                           StandardEnvironment environment, RocketMQTemplate rocketMqTemplate) {
        log.info("RocketMQListenerContainerConfig 初始化");

        this.rocketMQProperties = rocketMqProperties;
        this.rocketMQMessageConverter = rocketMqMessageConverter;
        this.rocketMqTemplate = rocketMqTemplate;
        this.environment = environment;
    }

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }

    /**
     * 实现BeanPostProcessor接口，在所有bean创建之后执行
     * @param bean bean对象
     * @param beanName bean名称
     * @return 返回入参的bean对象
     */
    @Override
    public Object postProcessAfterInitialization(@NonNull final Object bean, @NonNull final String beanName) {
        registerContainer(beanName, bean);
        return bean;
    }

    /**
     * 注册container，并生成consumer
     * @param beanName 使用@MqConsumer的Bean名称
     * @param bean 使用@MqConsumer的Bean
     */
    private void registerContainer(String beanName, Object bean) {
        if (!nonAnnotatedClasses.contains(bean.getClass())) {
            Class<?> targetClass = AopUtils.getTargetClass(bean);
            Map<Method, MqConsumer> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
                    (MethodIntrospector.MetadataLookup<MqConsumer>) method -> AnnotatedElementUtils.findMergedAnnotation(method, MqConsumer.class));

            if (annotatedMethods.isEmpty()) {
                // 过滤没有使用@MqConsumer的bean
                this.nonAnnotatedClasses.add(bean.getClass());
                log.debug("No @MqConsumer annotations found on bean type: " + bean.getClass());
            } else {
                for (Map.Entry<Method, MqConsumer> entry : annotatedMethods.entrySet()) {
                    Method method = entry.getKey();
                    MqConsumer mqConsumer = entry.getValue();

                    String containerBeanName = String.format("%s_%s", beanName, method.getName());

                    GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
                    // 注册container
                    genericApplicationContext.registerBean(containerBeanName, RocketMqListenerContainer.class,
                            () -> createContainer(containerBeanName, bean, method, mqConsumer));
                }
            }
        }
    }

    /**
     * 创建container，将需要创建consumer的参数传入到container里
     * @param name container名称
     * @param bean bean对象
     * @param method 使用@MqConsumer注解的方法
     * @param mqConsumer @MqConsumer 实例
     * @return container
     */
    private RocketMqListenerContainer createContainer(String name, Object bean, Method method, MqConsumer mqConsumer) {
        String consumerGroup = this.environment.resolvePlaceholders(mqConsumer.consumerGroup());
        String topic = this.environment.resolvePlaceholders(mqConsumer.topic());
        String tag = this.environment.resolvePlaceholders(mqConsumer.selectorExpression());

        RocketMqListenerContainer container = new RocketMqListenerContainer();
        container.setTopic(topic);
        container.setConsumerGroup(consumerGroup);
        container.setNameServer(rocketMQProperties.getNameServer());
        container.setName(name);
        container.setBean(bean);
        container.setMethod(method);
        container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
        container.setSelectorExpression(tag);
        container.setRocketMqTemplate(this.rocketMqTemplate);
        container.setCallbackTopic(mqConsumer.callbackTopic());
        return container;
    }
}
